package git.soulbgm.mapreduce.reduceJoin;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author SoulBGM
 * @date 2021/12/1 23:11
 */
public class TableReducer extends Reducer<Text, TableBean, TableBean, NullWritable> {

    private final String TABLE_NAME_ORDER = "order";
    private final String TABLE_NAME_PD = "pd";

    /*
    hadoop中迭代器的对象重用问题
    http://paddy-w.iteye.com/blog/1514595

    关于 hadoop 中 JVM 重用和对象重用的介绍
    http://wikidoop.com/wiki/Hadoop/MapReduce/Reducer
     */

    @Override
    protected void reduce(Text key, Iterable<TableBean> values, Context context) throws IOException, InterruptedException {
        String pname = "";
        List<TableBean> list = new ArrayList<>();
        for (TableBean value : values) {
            if (value.getTableName().equals(TABLE_NAME_PD)) {
                pname = value.getPname();
                break;
            } else {
                TableBean t = new TableBean();
                t.setId(value.getId());
                t.setPid(value.getPid());
                t.setAmount(value.getAmount());
                list.add(t);
            }
        }

        for (TableBean t : list) {
            t.setPname(pname);
            context.write(t, NullWritable.get());
        }
        for (TableBean value : values) {
            if (value.getTableName().equals(TABLE_NAME_ORDER)) {
                value.setPname(pname);
                context.write(value, NullWritable.get());
            }
        }
    }
}
